"""
Case Type   : DBlink功能
Case Name   : dblink连接最大数验证
Create At   : 2024/6/15
Owner       : peilinqian
Description :
    step1: 修改参数;
    step2: 重启数据库，使参数生效;
    step3: 创建数据库及库下创建dblink扩展;
    step4: 目标库进行数据库及用户建立;
    step5: 目标库进行pg_hba用户白名单配置;
    step6: 以线程开启事务，同时进行15个dblink连接;
    step7: 一个远程连接关闭后，再次进行2个连接;
Expect      :
    step1: 修改参数; expect:修改成功
    step2: 重启数据库，使参数生效; expect:重启成功
    step3: 创建数据库及库下创建dblink扩展; expect:成功
    step4: 目标库进行数据库及用户建立; expect:成功
    step5: 目标库进行pg_hba用户白名单配置; expect:成功
    step6: 以线程开启事务，同时进行15个dblink连接; expect:13个连接成功，2个失败
    step7: 一个远程连接关闭后，再次进行2个连接; expect:一个成功一个失败
History     :
"""
import os
import time
import unittest

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro

com = Common()


@unittest.skipIf(not com.check_node_exists('remote1_PrimaryDbuser'),
                 '无远程主机环境不执行')
class DblinkCase(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)}:初始化----')
        self.pri_dbuser = Node(node='PrimaryDbUser')
        self.pri_sh = CommonSH('PrimaryDbUser')
        self.remote_db = Node(node='remote1_PrimaryDbuser')
        self.remote_sh = CommonSH('remote1_PrimaryDbuser')
        self.com = Common('remote1_PrimaryDbuser')
        self.constant = Constant()
        self.db_name_1 = 'db_dblink_0022'
        self.u_sys = 'u_dblink_sys0022'
        self.err_flag = 'ERROR:  could not establish connection\n' \
                        'DETAIL:  FATAL:  Too many clients already, ' \
                        'current: 13, max_connections/reserved: 10/3'
        self.log.info('----查询参数----')
        self.guc1 = self.com.show_param("max_wal_senders",
                                        env_path=macro.DB_ENV_PATH_REMOTE1)
        self.guc2 = self.com.show_param("max_connections",
                                        env_path=macro.DB_ENV_PATH_REMOTE1)

    def test_main(self):
        step_txt = '----step1: 修改参数; expect:修改成功----'
        self.log.info(step_txt)
        msg = self.remote_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, f"max_wal_senders= 8",
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(msg, '执行失败:' + step_txt)
        msg = self.remote_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, f"max_connections= 10",
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(msg, '执行失败:' + step_txt)

        step_txt = '----step2: 重启数据库，使参数生效; expect:重启成功----'
        self.log.info(step_txt)
        restart_result = self.remote_sh.restart_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(restart_result)
        self.log.info('----查询参数----')
        self.com.show_param("max_wal_senders",
                            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.com.show_param("max_connections",
                            env_path=macro.DB_ENV_PATH_REMOTE1)

        step_txt = '----step3: 创建数据库及库下创建dblink扩展; expect:成功----'
        self.log.info(step_txt)
        db_sql = f"drop database if exists {self.db_name_1};" \
                 f"create database {self.db_name_1};"
        db_result = self.pri_sh.execut_db_sql(db_sql)
        self.log.info(db_result)
        create_sql = f"create extension dblink;"
        create_result = self.pri_sh.execut_db_sql(create_sql,
                                                  dbname=self.db_name_1)
        self.log.info(create_result)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, db_result,
                      "执行失败" + step_txt)
        self.assertIn(self.constant.create_extension_success, create_result,
                      "执行失败" + step_txt)

        step_txt = '----step4: 目标库进行数据库及用户建立; expect:成功----'
        self.log.info(step_txt)
        db_sql = f"drop database if exists {self.db_name_1};" \
                 f"create database {self.db_name_1};"
        msg1 = self.remote_sh.execut_db_sql(db_sql,
                                            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(msg1)
        user_sql = f"drop user if exists {self.u_sys} cascade;" \
                   f"create user {self.u_sys} sysadmin " \
                   f"password '{macro.PASSWD_INITIAL}';"
        msg2 = self.remote_sh.execut_db_sql(user_sql,
                                            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(msg2)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS, msg1,
                      "执行失败" + step_txt)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, msg2,
                      "执行失败" + step_txt)

        step_txt = '----step5: 目标库进行pg_hba用户白名单配置; expect:成功----'
        self.log.info(step_txt)
        self.log.info('----dblink端pg_hba配置----')
        guc_res = self.remote_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  {self.db_name_1}  all '
            f'{self.pri_dbuser.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, "执行失败" + step_txt)

        step6_txt = '----step6: 以线程开启事务，同时进行15个dblink连接; ' \
                    'expect:13个连接成功，2个失败----'
        self.log.info(step6_txt)
        session1 = []
        session1_result = []
        for dblink_item in range(15):
            if dblink_item == 1:
                sql = f"SELECT dblink_connect('dtest_{dblink_item}', " \
                      f"'host={self.remote_db.db_host} " \
                      f"port={self.remote_db.db_port} " \
                      f"dbname={self.db_name_1} " \
                      f"user={self.u_sys} " \
                      f"password={macro.PASSWD_INITIAL}');" \
                      f"select pg_sleep(10);"
            else:
                sql = f"SELECT dblink_connect('dtest_{dblink_item}', " \
                      f"'host={self.remote_db.db_host} " \
                      f"port={self.remote_db.db_port} " \
                      f"dbname={self.db_name_1} " \
                      f"user={self.u_sys} " \
                      f"password={macro.PASSWD_INITIAL}');" \
                      f"select pg_sleep(20);"
            session1.append(ComThread(self.pri_sh.execut_db_sql,
                                      args=(sql, '', self.db_name_1,)))
            session1[dblink_item].setDaemon(True)
            session1[dblink_item].start()

        time.sleep(11)
        step_txt = '----step7: 一个远程连接关闭后，再次进行2个连接; expect:一个成功一个失败----'
        self.log.info(step_txt)
        session2 = []
        session2_result = []
        for dblink_item in range(2):
            sql = f"SELECT dblink_connect('dtest1_{dblink_item}', " \
                  f"'host={self.remote_db.db_host} " \
                  f"port={self.remote_db.db_port} " \
                  f"dbname={self.db_name_1} " \
                  f"user={self.u_sys} " \
                  f"password={macro.PASSWD_INITIAL}');" \
                  f"select pg_sleep(10);"
            session2.append(ComThread(self.pri_sh.execut_db_sql,
                                      args=(sql, '', self.db_name_1,)))
            session2[dblink_item].setDaemon(True)
            session2[dblink_item].start()
        for dblink_item in range(2):
            session2[dblink_item].join()
            session2_result.append(session2[dblink_item].get_result())
        self.log.info(session2_result)
        ok_list = [item for item in session2_result if 'OK' in item]
        self.assertEqual(1, len(ok_list), '执行失败:' + step6_txt)
        err_list = [item for item in session2_result if self.err_flag in item]
        self.assertEqual(1, len(err_list), '执行失败:' + step6_txt)

        step_txt = '----获取步骤6执行结果;----'
        self.log.info(step_txt)
        for dblink_item in range(15):
            session1[dblink_item].join()
            session1_result.append(session1[dblink_item].get_result())
        self.log.info(session1_result)
        ok_list = [item for item in session1_result if 'OK' in item]
        self.assertEqual(13, len(ok_list), '执行失败:' + step6_txt)
        err_list = [item for item in session1_result if self.err_flag in item]
        self.assertEqual(2, len(err_list), '执行失败:' + step6_txt)

    def tearDown(self):
        self.log.info('----this is tearDown----')
        step8_txt = '----dblink侧清理数据; expect:清理成功----'
        self.log.info(step8_txt)
        drop_sql = f"drop database {self.db_name_1};" \
                   f"drop user if exists {self.u_sys} cascade;"
        msg1 = self.pri_sh.execut_db_sql(drop_sql)
        self.log.info(msg1)

        step9_txt = '----目标库进行数据库及用户删除; expect:成功----'
        self.log.info(step9_txt)
        drop_db_sql = f"clean connection to all force for database " \
                      f"{self.db_name_1};" \
                      f"drop database if exists {self.db_name_1};"
        drop_db_msg = self.remote_sh.execut_db_sql(
            drop_db_sql, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_db_msg)
        drop_user_sql = f"drop user if exists {self.u_sys} cascade;"
        drop_user_msg = self.remote_sh.execut_db_sql(
            drop_user_sql, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_user_msg)

        step10_txt = '----目标库进行pg_hba取消用户白名单配置; expect: 成功----'
        self.log.info(step10_txt)
        guc_res1 = self.remote_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  {self.db_name_1}  all '
            f'{self.pri_dbuser.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res1)

        step11_txt = '----目标用户侧还原参数; expect:修改成功----'
        self.log.info(step11_txt)
        msg2 = self.remote_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"max_wal_senders= {self.guc1}",
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        msg3 = self.remote_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"max_connections= {self.guc2}",
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        step_txt = '----重启数据库，使参数生效; expect:重启成功----'
        self.log.info(step_txt)
        restart_result = self.remote_sh.restart_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info('----查询参数----')
        self.com.show_param("max_wal_senders",
                            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.com.show_param("max_connections",
                            env_path=macro.DB_ENV_PATH_REMOTE1)

        self.log.info('----teardown断言----')
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, msg1,
                      "执行失败" + step8_txt)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, msg1,
                      "执行失败" + step8_txt)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS, drop_db_msg,
                      "执行失败" + step9_txt)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, drop_user_msg,
                      "执行失败" + step9_txt)

        self.assertTrue(guc_res1, "执行失败" + step10_txt)
        self.assertTrue(msg2, '执行失败:' + step11_txt)
        self.assertTrue(msg3, '执行失败:' + step11_txt)
        self.assertTrue(restart_result, '执行失败:' + step11_txt)

        self.log.info(f'----{os.path.basename(__file__)}:执行完毕----')
